from kafka import KafkaProducer
import json

# 公共配置
BOOTSTRAP_SERVERS = 'localhost:9092'

def produce_txza_xzrk(
    GMSFHM: str = "123456789",
    XM: str = "张三",
    bootstrap_servers: str = BOOTSTRAP_SERVERS
):
    """生产新型人口数据到 txza_xzrk Topic"""
    data = {
        "GMSFHM": GMSFHM,
        "XM": XM
    }
    _send_message('txza_xzrk', data, bootstrap_servers)

def produce_txza_zzrk(
    SFZH: str = "123456789",
    XM: str = "张三",
    bootstrap_servers: str = BOOTSTRAP_SERVERS
):
    """生产暂住人口数据到 txza_zzrk Topic"""
    data = {
        "SFZH": SFZH,
        "XM": XM
    }
    _send_message('txza_zzrk', data, bootstrap_servers)

def produce_txqb_rkzp(
    SFZH: str = "123456789",
    URL: str = "http://192.168.2.242/test1.jpg",
    bootstrap_servers: str = BOOTSTRAP_SERVERS
):
    """生产人口照片数据到 txqb_rkzp Topic"""
    data = {
        "SFZH": SFZH,
        "URL": URL
    }
    _send_message('txqb_rkzp', data, bootstrap_servers)

def _send_message(topic: str, data: dict, bootstrap_servers: str):
    """内部通用发送方法"""
    producer = None
    try:
        producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda x: json.dumps(x).encode('utf-8'),
            acks='all'
        )
        producer.send(topic, value=data)
        producer.flush()
        print(f"成功发送数据到 {topic}: {json.dumps(data, ensure_ascii=False)}")
    except Exception as e:
        print(f"发送到 {topic} 失败: {str(e)}")
    finally:
        if producer:
            producer.close()

# 使用示例 --------------------------
if __name__ == "__main__":
    # 发送默认测试数据
    produce_txza_xzrk()
    produce_txza_zzrk()
    produce_txqb_rkzp()

    # 发送自定义数据示例

